package cn.fastoo.task;

import cn.fastoo.common.utils.HttpClientHelper;
import cn.fastoo.common.utils.RegexUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * DR 推送
 */
public class DrQueue {

    private final static Logger logger = LoggerFactory.getLogger(DrQueue.class);


    private static BlockingQueue<DrMessage> blockingQueue = new LinkedBlockingQueue<DrMessage>();


    static {
        Consumer consumer=new Consumer(blockingQueue);

        new Thread(consumer,"dr_queue").start();

    }

    public static void put(DrMessage drMessage){
        try {
            blockingQueue.put(drMessage);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //消费者
    public static class Consumer implements Runnable {

        private final BlockingQueue<DrMessage> blockingQueue;

        private volatile boolean flag;

        public Consumer(BlockingQueue<DrMessage> blockingQueue) {
            this.blockingQueue = blockingQueue;
        }

        public void run() {
            while(!flag){

                try {
                    DrMessage drMessage = blockingQueue.take();

                    String da = drMessage.getDa();
                    String fastooMsgid = drMessage.getFastooMsgid();
                    String drUrl = drMessage.getDrUrl();
                    Integer status = drMessage.getStatus(); //0失败1成功
                    Integer retry = drMessage.getRetry();

                    if (StringUtils.isNotBlank(drUrl)&& RegexUtils.checkURL(drUrl)) {

                        try {
                            String drUrlReq = drUrl+"?msgid="+fastooMsgid+"&da="+da+"&status="+status;
                            CloseableHttpClient client = HttpClientHelper.getHttpClient();
                            HttpGet httpGet = new HttpGet(drUrlReq);
                            CloseableHttpResponse httpResponse = client.execute(httpGet);
                            int statusCode = httpResponse.getStatusLine().getStatusCode();
                            EntityUtils.consume(httpResponse.getEntity());
                            logger.info("DR[" + drUrl + "]Result：" + statusCode);

                            httpResponse.close();
                        } catch (Exception e) {
                            try {
                                logger.error("DR请求[" + drUrl + "]出错：", e);
                                if (retry<=3) {
                                    Thread.sleep(10000);
                                    drMessage.setDa(da);
                                    drMessage.setStatus(status);
                                    drMessage.setFastooMsgid(fastooMsgid);
                                    drMessage.setDrUrl(drUrl);
                                    drMessage.setRetry(retry+1);
                                    DrQueue.put(drMessage);
                                }
                            } catch (Exception ex) {
                                logger.error("dr请求重发失败", ex);
                            }
                        }
                    }
                } catch (Exception e) {
                    logger.error("处理DR异常:"+e);
                }
            }
        }
        public void shutDown(){
            flag=true;
        }
    }





    public static void main(String[] args) throws InterruptedException {

    }
}
